package com.taobao.api.internal.tmc;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.taobao.api.internal.util.NamedThreadFactory;
import com.taobao.api.internal.util.StringUtils;
import com.taobao.api.internal.util.TaobaoUtils;
import com.taobao.api.internal.util.json.JSONReader;
import com.taobao.api.internal.util.json.JSONValidatingReader;
import com.taobao.top.link.LinkException;
import com.taobao.top.link.endpoint.EndpointContext;
import com.taobao.top.link.endpoint.Identity;

/**
 * 消息通道客户端入口。
 * 
 * @author fengsheng
 * @since 1.0, May 4, 2013
 */
public class TmcClient {

	private static final Log log = LogFactory.getLog(TmcClient.class);

	// data transfer protocol
	private static final String SIZE = "size";
	private static final String MSG_LIST = "msg_list";
	private static final String SUCCESS = "success";
	private static final String FAIL = "fail";
	private static final String CONTENT = "content";
	private static final String SESSION = "session";
	private static final String TIMESTAMP = "timestamp";
	private static final String APP_KEY = "app_key";
	private static final String GROUP_NAME = "group_name";
	private static final String SIGN = "sign";

	protected static final String KIND = "__kind";
	protected static final String TOPIC = "__topic";

	// just different from type
	public class Kind {
		public static final short Data = 0;
		public static final short Confirm = 1;
		public static final short Request = 2;
		public static final short Reply = 3;
	}

	private final AtomicBoolean connected = new AtomicBoolean(false);
	private final ReentrantLock fetchLock = new ReentrantLock();

	private InnerClient client;
	private MessageHandler handler;

	private ThreadPoolExecutor threadPool;
	private int threadCount = 20; // 并发处理的线程数量
	private int fetchSize = 100; // 每次批量获取的消息数量
	private int fetchPeriod = 10; // 定时获取消息周期（单位：秒）

	private Timer fetchTimer;
	private TimerTask fetchTimerTask;
	private boolean fetchStopFlag;

	private String uri; // 消息通道服务地址

	public TmcClient(String appKey, String appSecret) {
		this(appKey, appSecret, "default"); // 默认分组+线上服务
	}

	public TmcClient(String appKey, String appSecret, String groupName) {
		this("ws://mc.api.taobao.com/", appKey, appSecret, groupName); // 指定分组+线上服务
	}

	public TmcClient(String uri, String appKey, String appSecret, String groupName) {
		this.uri = uri;
		this.client = new InnerClient(new TmcIdentity(appKey, groupName));
		this.client.appKey = appKey;
		this.client.appSecret = appSecret;
		this.client.groupName = groupName;
	}

	public void setMessageHandler(MessageHandler handler) {
		this.handler = handler;
	}

	public void setThreadCount(int threadCount) {
		this.threadCount = threadCount;
	}

	public void setFetchSize(int fetchSize) {
		this.fetchSize = fetchSize;
	}

	public void setFetchPeriod(int fetchPeriod) {
		this.fetchPeriod = fetchPeriod;
	}

	public void connect() throws LinkException {
		if (!connected.compareAndSet(false, true)) {
			return;
		}

		if (this.handler != null) {
			this.client.setMessageHandler(new InnerMessageHandler());
			this.threadPool = new ThreadPoolExecutor(threadCount, threadCount, fetchPeriod, TimeUnit.MICROSECONDS,
					new ArrayBlockingQueue<Runnable>(fetchSize), new NamedThreadFactory("tmc-exe"), new CallerRunsPolicy());
		}
		this.client.connect(uri);
		this.doFetch();
	}

	/**
	 * 向指定的主题发布一条与用户相关的消息。
	 * 
	 * @param topic 主题名称
	 * @param content 严格根据主题定义的消息内容（JSON/XML）
	 * @param session 用户授权码
	 */
	public void send(String topic, String content, String session) throws LinkException {
		if (StringUtils.isEmpty(content)) {
			throw new LinkException("content is required");
		}
		if (StringUtils.isEmpty(session)) {
			throw new LinkException("session is required");
		}

		Map<String, String> msg = new HashMap<String, String>();
		msg.put(KIND, Short.toString(Kind.Data));
		msg.put(TOPIC, topic);
		msg.put(CONTENT, content);
		msg.put(SESSION, session);
		this.client.sendAndWait(msg, 1000);
	}

	public void close() {
		this.stopFetch();
		if (this.threadPool != null) {
			this.threadPool.shutdown();
		}
		this.client.disconnect();
	}

	private void doFetch() {
		this.stopFetch();
		this.fetchTimerTask = new TimerTask() {
			public void run() {
				if (!fetchStopFlag && !fetchLock.isLocked()) {
					fetch();
				}
				fetchStopFlag = false;
			}
		};
		Date begin = new Date();
		begin.setTime(begin.getTime() + fetchPeriod * 1000L);
		this.fetchTimer = new Timer(true);
		this.fetchTimer.schedule(this.fetchTimerTask, begin, fetchPeriod * 1000L);
	}

	private void stopFetch() {
		if (this.fetchTimer != null) {
			this.fetchTimer.cancel();
			this.fetchTimer = null;
		}
	}

	private void fetch() {
		Map<String, String> msg = new HashMap<String, String>();
		msg.put(KIND, Short.toString(Kind.Request));
		msg.put(SIZE, Integer.toString(this.fetchSize));
		try {
			this.client.send(msg);
		} catch (LinkException e) {
			log.error("fetch message error", e);
		}
	}

	private void delayNextFetch() {
		this.fetchStopFlag = true;
	}

	class InnerClient extends MixClient {
		private String appKey;
		private String appSecret;
		private String groupName;

		public InnerClient(TmcIdentity id) {
			super(id);
		}

		@Override
		protected Map<String, String> createConnectHeaders() {
			Map<String, String> headers = new HashMap<String, String>();
			headers.put(TIMESTAMP, String.valueOf(System.currentTimeMillis()));
			headers.put(APP_KEY, this.appKey);
			headers.put(GROUP_NAME, this.groupName);
			String sign = null;
			try {
				sign = TaobaoUtils.signTopRequestNew(headers, this.appSecret, false);
			} catch (Exception e) {
				log.error("sign error", e);
			}
			headers.put(SIGN, sign);
			return headers;
		}
	}

	class InnerMessageHandler implements com.taobao.top.link.endpoint.MessageHandler {
		// on push messages
		public void onMessage(EndpointContext context) throws Exception {
			this.onMessage(context.getMessage(), context.getMessageFrom());
		}

		// on poll results
		public void onMessage(Map<String, String> message, Identity identity) {
			fetchLock.lock();
			try {
				String body = message.get(MSG_LIST);
				List<Message> msgList = parseJsonMessages(body);
				final CountDownLatch latch = new CountDownLatch(msgList.size());

				final Map<String, String> confirm = new HashMap<String, String>();
				confirm.put(KIND, Short.toString(Kind.Confirm));

				 // using string buffer to make sure thread-safe
				final StringBuffer sIds = new StringBuffer();
				final StringBuffer fIds = new StringBuffer();
				for (final Message msg : msgList) {
					threadPool.submit(new Runnable() {
						public void run() {
							try {
								MessageStatus status = new MessageStatus();
								handler.onMessage(msg, status);
								if (status.isFail()) {
									appendMsgId(fIds, msg.getId());
								} else {
									appendMsgId(sIds, msg.getId());
								}
							} catch (Throwable e) {
								appendMsgId(fIds, msg.getId());
								log.error("handle message fail: msgId=" + msg.getId(), e);
							} finally {
								latch.countDown();
							}
						}
					});
				}

				latch.await();
				// batch confirm message handle result
				if (sIds.length() > 0) {
					confirm.put(SUCCESS, sIds.toString());
				}
				if (fIds.length() > 0) {
					confirm.put(FAIL, fIds.toString());
				}
				client.send(confirm);

				// immediately fetch messages only if confirmed
				fetch();
				delayNextFetch();
			} catch (InterruptedException e) {
				Thread.currentThread().interrupt();
			} catch (LinkException e) {
				log.error("confirm message error", e);
			} catch (Throwable e) {
				log.error("handle message error", e);
			} finally {
				fetchLock.unlock();
			}
		}

		private void appendMsgId(StringBuffer ids, Long id) {
			if (ids.length() > 0) {
				ids.append(",");
			}
			ids.append(id);
		}

		/**
		 * 解释JSON消息成对象，消息格式：[{msg},{msg}]
		 */
		private List<Message> parseJsonMessages(String json) {
			List<Message> msgList = new ArrayList<Message>();

			JSONReader reader = new JSONValidatingReader();
			Object root = reader.read(json);
			if (root instanceof List<?>) {
				List<?> list = (List<?>) root;
				for (Object obj : list) {
					if (obj instanceof Map<?, ?>) {
						Map<?, ?> map = (Map<?, ?>) obj;
						msgList.add(Message.parse(map));
					}
				}
			}

			return msgList;
		}
	}

}
